package cloud.xiguapi.ubas.analysis.hotpages;

import cloud.xiguapi.ubas.analysis.hotpages.model.*;
import cloud.xiguapi.ubas.model.ApacheLogEvent;
import cloud.xiguapi.ubas.model.PageViewCount;
import cloud.xiguapi.ubas.utils.JacksonUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.text.SimpleDateFormat;
import java.util.regex.Pattern;

import static cloud.xiguapi.ubas.constant.JobNameConstant.FILE_HOT_PAGES;

/**
 * 离线热门页面统计
 *
 * @author 大大大西西瓜皮🍉
 * date: 2021-5-18 下午 10:36
 * desc:
 */
@Component
public class FileHotPagesAnalysis {

    /**
     *
     */
    public static final String RESOURCE_FILE_REG = "^((?!\\.(css|js|png|ico|jpg|jpeg)$).)*$";

    @Resource(name = "env")
    private StreamExecutionEnvironment env;

    @Async("fileHotPagesAnalysis")
    public void analysis(String path) throws Exception {
        DataStream<String> inputDataStream = env.readTextFile(path);
        DataStream<ApacheLogEvent> logEventDataStream = inputDataStream.map(line -> {
            String[] fields = line.split(" ");

            // 转换时间戳
            SimpleDateFormat format = new SimpleDateFormat("dd/MM/yyy:HH:mm:ss");
            Long timestamp = format.parse(fields[3]).getTime();

            // 封装实体
            return ApacheLogEvent.builder()
                    .ip(fields[0])
                    .userId(fields[1])
                    .timestamp(timestamp)
                    .method(fields[5])
                    .url(fields[6])
                    .build();
        }).assignTimestampsAndWatermarks(new HotPagesBoundedOutOfOrdernessTimestampExtractor(Time.seconds(1L)));

        // logEventDataStream.print("log event");

        // 定义一个侧输出流标签
        // OutputTag<ApacheLogEvent> lateTag = new OutputTag<>("late");

        // 分组开窗聚合
        SingleOutputStreamOperator<PageViewCount> windowAggregateDataStream = logEventDataStream
                // 只选择GET请求
                .filter(data -> "GET".equals(data.getMethod()))
                // 过滤资源文件
                .filter(data -> Pattern.matches(RESOURCE_FILE_REG, data.getUrl()))
                // 按照URL分组
                .keyBy(ApacheLogEvent::getUrl)
                // 开窗
                .timeWindow(Time.minutes(10L), Time.seconds(5L))
                // 允许迟到数据, 延迟数据的第三重保证, 确保数据的完整性
                .allowedLateness(Time.minutes(1L))
                // 添加侧输出流
                // .sideOutputLateData(lateTag)
                // 聚合, 传入增量聚合函数对象和全窗口函数对象
                .aggregate(new FileHotPageCountAggregate(), new FileHotPageCountResult());

        windowAggregateDataStream.print("agg");
        // windowAggregateDataStream.getSideOutput(lateTag).print("late");


        // 收集同一窗口count数据, 排序输出
        DataStream<TopNHotPagesResult> resultDataStream = windowAggregateDataStream
                .keyBy(PageViewCount::getWindowEnd)
                .process(new TopNHotPages(5));

        resultDataStream.print();
        resultDataStream.addSink(new FileHotPagesMySqlSink());

        env.execute(FILE_HOT_PAGES.getJobName());
    }

    public static class FileHotPagesMySqlSink extends RichSinkFunction<TopNHotPagesResult> {

        private transient Connection connection;

        private transient PreparedStatement clearStatement;

        private transient PreparedStatement selectStatement;

        private transient PreparedStatement insertStatement;

        private transient PreparedStatement updateStatement;

        private transient boolean isFirst = true;

        @Override
        public void open(Configuration parameters) throws Exception {
            connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/ubas?serverTimezone=UTC", "root", "1015");
            clearStatement = connection.prepareStatement("TRUNCATE TABLE file_hot_pages");
            selectStatement = connection.prepareStatement("SELECT 1 FROM file_hot_pages WHERE window_end = ? LIMIT 1");
            insertStatement = connection.prepareStatement("INSERT INTO file_hot_pages(window_end, details) VALUES(?, ?)");
            updateStatement = connection.prepareStatement("UPDATE file_hot_pages SET details = ? WHERE window_end = ?");
        }

        @Override
        public void close() throws Exception {
            updateStatement.close();
            selectStatement.close();
            clearStatement.close();
            insertStatement.close();
            connection.close();
        }

        @Override
        public void invoke(TopNHotPagesResult value, SinkFunction.Context context) throws Exception {
            if (isFirst) {
                clearStatement.execute();
                isFirst = false;
                return;
            }

            selectStatement.setString(1, value.getWindowEnd());
            boolean next = selectStatement.executeQuery().next();

            if (next) {
                updateStatement.setString(1, JacksonUtils.obj2json(value.getDetails()));
                updateStatement.setString(2, value.getWindowEnd());
                updateStatement.execute();
            } else {
                insertStatement.setString(1, value.getWindowEnd());
                insertStatement.setString(2, JacksonUtils.obj2json(value.getDetails()));
                insertStatement.execute();
            }
        }
    }
}
